package com.xmh.code.mq.consumer;

import com.xmh.code.entity.PingRecordEntity;
import com.xmh.code.repository.PingRecordRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.time.LocalDateTime;

@Slf4j
@Service
@RocketMQMessageListener(topic = "${mq-config.topic}", consumerGroup = "${rocketmq.consumer.group}", messageModel = MessageModel.CLUSTERING)
public class RocketMQConsumer implements RocketMQListener<String> {

    public static final String FILE_PATH = "shared_file.txt";

    @Resource
    private WebClient webClient;
    @Resource
    private PingRecordRepository repository;

    @Value("${pong.api}")
    public String pongServerApi;

    @Override
    public void onMessage(String message) {
        try (FileChannel channel = FileChannel.open(Paths.get(FILE_PATH), StandardOpenOption.WRITE);
             FileLock fileLock = channel.tryLock()) {
            if (fileLock != null) {
                send(message);
            } else {
                this.saveNotSendRecord(message);
                log.info("【{}】get file lock one failed", message);
            }
        } catch (Exception e) {
            this.saveNotSendRecord(message);
            log.info("【{}】get file lock one error", message);
        }
    }

    private void saveNotSendRecord(String message) {
        PingRecordEntity pingRecordEntity = new PingRecordEntity();
        pingRecordEntity.setRequestContent("hello");
        pingRecordEntity.setPingServerName(message);
        pingRecordEntity.setSuccess(false);
        pingRecordEntity.setDate(LocalDateTime.now().plusHours(8));
        pingRecordEntity.setSend(false);
        this.repository.save(pingRecordEntity);
    }

    public void send(String name) throws InterruptedException {
        log.info("【{}】Start ping pong ===========================", name);
        PingRecordEntity pingRecordEntity = new PingRecordEntity();
        pingRecordEntity.setRequestContent("hello");
        pingRecordEntity.setPingServerName(name);
        pingRecordEntity.setSuccess(true);
        pingRecordEntity.setDate(LocalDateTime.now().plusHours(8));
        pingRecordEntity.setHttpStatusCode(200);
        pingRecordEntity.setSend(true);

        this.webClient.get()
                .uri(this.pongServerApi + name)
                .retrieve()
                .onStatus(status -> status.value() != 200, clientResponse -> {
                    pingRecordEntity.setSuccess(false);
                    pingRecordEntity.setHttpStatusCode(clientResponse.statusCode().value());
                    this.repository.save(pingRecordEntity);
                    return Mono.empty();
                })
                .bodyToMono(String.class).subscribe(response -> {
                    pingRecordEntity.setResponseContent(response);
                    this.repository.save(pingRecordEntity);
                    log.info("【{}】Received response: {}", name, response);
                });
        Thread.sleep(500);
    }
}
